package com.renrendoc.network.flow.ana;

import com.renrendoc.network.flow.beans.out.TopNPage;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.shaded.guava30.com.google.common.collect.Lists;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;

import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.Map;

public class TopNHotPages extends KeyedProcessFunction<Long, TopNPage, ArrayList<TopNPage>> {
        private Integer topSize;

        public TopNHotPages(Integer topSize) {
            this.topSize = topSize;
        }

        // 定义状态，保存当前所有TopNPage到Map中
//        ListState<TopNPage> TopNPageListState;
        MapState<Long, Long> TopNPageMapState;

        @Override
        public void open(Configuration parameters) throws Exception {
//            TopNPageListState = getRuntimeContext().getListState(new ListStateDescriptor<TopNPage>("page-count-list", TopNPage.class));
            TopNPageMapState = getRuntimeContext().getMapState(new MapStateDescriptor<Long, Long>("page-count-map", Long.class, Long.class));
        }

        @Override
        public void processElement(TopNPage value, Context ctx, Collector<ArrayList<TopNPage>> out) throws Exception {
//            TopNPageListState.add(value);
            TopNPageMapState.put(value.aid, value.total);
            long t = System.currentTimeMillis();
            ctx.timerService().registerEventTimeTimer(t);
            // 注册一个1分钟之后的定时器，用来清空状态
            ctx.timerService().registerEventTimeTimer(t + 60 * 1000L);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<ArrayList<TopNPage>> out) throws Exception {
            // 先判断是否到了窗口关闭清理时间，如果是，直接清空状态返回
            if( timestamp >= ctx.getCurrentKey() + 60 * 1000L ){
                TopNPageMapState.clear();
                return;
            }

            Iterable<Map.Entry<Long, Long>> entries = TopNPageMapState.entries();
            ArrayList<Map.Entry<Long, Long>> TopNPages = Lists.newArrayList(entries);

            TopNPages.sort(new Comparator<Map.Entry<Long, Long>>() {

                @Override
                public int compare(Map.Entry<Long, Long> o1, Map.Entry<Long, Long> o2) {
                    if(o1.getValue() > o2.getValue())
                        return -1;
                    else if(o1.getValue() < o2.getValue())
                        return 1;
                    else
                        return 0;
                }
            });

            // 格式化成String输出
            ArrayList<TopNPage> topNPages = new ArrayList<>();
            StringBuilder resultBuilder = new StringBuilder();
            resultBuilder.append("===================================\n");
            // 遍历列表，取top n输出
            for (int i = 0; i < Math.min(topSize, TopNPages.size()); i++) {
                Map.Entry<Long, Long> currentItemViewCount = TopNPages.get(i);
                resultBuilder.append("NO ").append(i + 1).append(":")
                        .append(" 页面URL = ").append(currentItemViewCount.getKey())
                        .append(" 浏览量 = ").append(currentItemViewCount.getValue())
                        .append("\n");
                TopNPage topNPage = new TopNPage(currentItemViewCount.getKey(), currentItemViewCount.getValue(), ctx.getCurrentKey());
                topNPage.ident = "book";
                topNPages.add(topNPage);
            }
            resultBuilder.append("===============================\n\n");

            // 控制输出频率
            System.out.println(resultBuilder);

            out.collect(topNPages);

//            TopNPageListState.clear();
        }
    }